package cn.doitedu.rtdw.demos;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/6
 * @Desc: 学大数据，到多易教育
 *  利用定时器延迟重试查询hbase维度数据的示例
 **/
public class TimerRetryQueryHbase {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        DataStreamSource<String> userNameStream = env.socketTextStream("doitedu", 4444);

        KeyedStream<String, String> keyed = userNameStream.keyBy(name -> name);

        SingleOutputStreamOperator<String> result = keyed.process(new KeyedProcessFunction<String, String, String>() {
            Table testTable;
            MapState<Long, String> timers;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 创建hbase的客户端连接
                org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
                hbaseConf.set("hbase.zookeeper.quorum", "doitedu:2181");
                Connection conn = ConnectionFactory.createConnection(hbaseConf);
                testTable = conn.getTable(TableName.valueOf("test"));

                // 记录所注册的定时器信息：  定时器触发时间，要查询的人
                timers = getRuntimeContext().getMapState(new MapStateDescriptor<Long, String>("retries", Long.class, String.class));

            }

            @Override
            public void processElement(String userName, KeyedProcessFunction<String, String, String>.Context ctx, Collector<String> out) throws Exception {

                Get get = new Get(userName.getBytes());
                Result result = testTable.get(get);
                if (result.isEmpty()) {

                    long timerTime = ctx.timerService().currentProcessingTime() + 5000;  // 获取此刻的处理时间  +  要延迟的时间 ==> 要重试的时间点

                    ctx.timerService().registerProcessingTimeTimer(timerTime);  // 注册定时器，延迟重试
                    timers.put(timerTime, userName); // 将注册的定时器信息，记录到状态中

                } else {
                    byte[] phoneBytes = result.getValue("f".getBytes(), "phone".getBytes());
                    String phone = Bytes.toString(phoneBytes);
                    out.collect(userName + "," + phone);
                }
            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, String, String>.OnTimerContext ctx, Collector<String> out) throws Exception {

                String userName = timers.get(timestamp);
                Get get = new Get(userName.getBytes());
                Result result = testTable.get(get);

                if (result.isEmpty()) {
                    // 注册定时器，延迟重试
                    long timerTime = ctx.timerService().currentProcessingTime() + 5000;
                    ctx.timerService().registerProcessingTimeTimer(timerTime);
                    timers.put(timerTime, userName);

                } else {
                    byte[] phoneBytes = result.getValue("f".getBytes(), "phone".getBytes());
                    String phone = Bytes.toString(phoneBytes);
                    out.collect(userName + "," + phone);
                }

            }
        });


        result.print();

        env.execute();

    }
}
